﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;

using ZK = ZooKeeperNet;
using Org.Apache.Zookeeper.Data;

using WorkStation.Frameworks.Core;
using WorkStation.Frameworks.Core.Extensions;
using WorkStation.Frameworks.EasyBus.DbModel;
using WorkStation.Frameworks.EasyBus.Repositories;

using Newtonsoft.Json;

namespace WorkStation.Frameworks.EasyBus.Repositories.ZooKeeper
{
    public class RegisterCenterRepository : IRegisterCenterRepository
    {
        protected string _connectionString = string.Empty;
        protected ZK.ZooKeeper _zk = null;

        protected IBus _bus = null;

        protected static readonly string ROUTEKEY_PREFIX = "{RouteKey}";
        protected static readonly string PUBBUSNAME_PREFIX = "{PubBusName}";
        protected static readonly string NODENAME_PREFIX = "{NodeName}";
        protected static readonly string SUBBUSNAME_PREFIX = "{SubBusName}";
        protected static readonly string SUBSCRIBE_NAME_PREFIX = "{SubscribeName}";
        protected static readonly string LISTENER_NAME_PREFIX = "{ListenerName}";

        protected static readonly string ROOT = "/EasyBus";

        protected static readonly string PUBLISH = "{0}/PublishInfo".Fmt(ROOT);
        protected static readonly string PUBLISH_BUSNAME_NODE = "{0}/{1}".Fmt(PUBLISH, PUBBUSNAME_PREFIX);
        protected static readonly string PUBLISH_ROUTEKEY_NODE = "{0}/{1}".Fmt(PUBLISH_BUSNAME_NODE, ROUTEKEY_PREFIX);
        protected static readonly string PUBLISH_SERVER_NODE = "{0}/{1}".Fmt(PUBLISH_ROUTEKEY_NODE, NODENAME_PREFIX);

        protected static readonly string SUBSCRIBE = "{0}/SubscribeInfo".Fmt(ROOT);
        protected static readonly string SUBSCRIBE_PUBBUSNAME_NODE = "{0}/{1}".Fmt(SUBSCRIBE, PUBBUSNAME_PREFIX);
        protected static readonly string SUBSCRIBE_ROUTEKEY_NODE = "{0}/{1}".Fmt(SUBSCRIBE_PUBBUSNAME_NODE, ROUTEKEY_PREFIX);
        protected static readonly string SUBSCRIBE_NODE = "{0}/{1}".Fmt(SUBSCRIBE_ROUTEKEY_NODE, SUBSCRIBE_NAME_PREFIX);

        protected static readonly string LISTENER = "{0}/ListenerInfo".Fmt(ROOT);
        protected static readonly string LISTENER_SUBBUSNAME_NODE = "{0}/{1}".Fmt(LISTENER, SUBBUSNAME_PREFIX);
        protected static readonly string LISTENER_SERVER_NODE = "{0}/{1}".Fmt(LISTENER_SUBBUSNAME_NODE, NODENAME_PREFIX);
        protected static readonly string LISTENER_NAME_NODE = "{0}/{1}".Fmt(LISTENER_SERVER_NODE, LISTENER_NAME_PREFIX);
        
        public RegisterCenterRepository(string connectionString, IBus bus, TimeSpan sessionTimeout) 
        {
            connectionString.IsNullOrEmpty().WhenTrue(() => { throw new ArgumentNullException("connectionString"); });

            this._connectionString = connectionString;

            this._bus = bus;
            this._zk = this.GetZK(sessionTimeout);
            
            this.InitZooKeeperNode();
        }

        #region IRegisterCenterRepository 成员

        public List<PublishInfo> GetBusPublishInfo(string busName)
        {
            throw new NotImplementedException();
        }

        public List<ListenerInfo> GetBusListenerInfo(string busName)
        {
            //ListenerInfo Path:/EasyBus/ListenerInfo/{SubBusName}/{ServerNode}/{Mq}

            List<ListenerInfo> result = new List<ListenerInfo>();

            string busNameNodePath = GetListenerSubBusNameNodePath(busName);

            (this._zk.Exists(busNameNodePath, false) != null).WhenTrue(() =>
            {
                foreach (string nodeName in this._zk.GetChildren(busNameNodePath, false))
                {
                    string nodeNamePath = GetListenerServerNodePath(busName, nodeName);

                    foreach (string listenerName in this._zk.GetChildren(nodeNamePath, false))
                    {
                        string listenerNodePath = GetListenerNameNodePath(busName, nodeName, listenerName);

                        Stat stat = this._zk.Exists(listenerNodePath, false);

                        if (stat != null)
                        {
                            byte[] data = this._zk.GetData(listenerNodePath, new ListenerInfoChangeWatcher(this._bus, this._zk), stat);

                            if (data != null)
                            {
                                result.Add(JsonConvert.DeserializeObject<ListenerInfo>(data.ToStringByUTF8()));
                            }
                        }
                    }
                }
            });

            return result;
        }

        public List<ListenerInfo> GetBusListenerInfoByNode(string busName, string nodeName) 
        {
            return this.GetBusListenerInfo(busName).Where(v => v.NodeName.Equals(nodeName)).ToList();
        }

        public List<SubscribeInfo> GetBusSubscribeInfo(string busName)
        {
            List<SubscribeInfo> result = new List<SubscribeInfo>();

            try
            {
                string publishBusNodePath = GetSubscribePubBusNameNodePath(busName);

                if (this._zk.Exists(publishBusNodePath, false) != null)
                {
                    foreach (string routeKey in this._zk.GetChildren(publishBusNodePath, false))
                    {
                        string routeKeyNodePath = GetSubscribeRouteKeyNodePath(busName, routeKey);

                        foreach (string subscribeName in this._zk.GetChildren(routeKeyNodePath, new SubscribeInfoChangeWatcher(this._bus, this._zk)))
                        {
                            string subscribeNameNodePath = GetSubscribeNameNodePath(busName, routeKey, subscribeName);

                            Stat stat = this._zk.Exists(subscribeNameNodePath, false);

                            if (stat != null)
                            {
                                byte[] data = this._zk.GetData(subscribeNameNodePath, new SubscribeInfoChangeWatcher(this._bus, this._zk), stat);

                                if (data != null)
                                {
                                    result.Add(JsonConvert.DeserializeObject<SubscribeInfo>(data.ToStringByUTF8()));
                                }
                            }
                        }
                    }
                }
            }
            catch (Exception ce) 
            {
                throw new EasyBusException("IRegisterCenterRepository.GetBusSubscribeInfo()方法出错，详细见内部异常", ce);
            }

            return result;
        }

        public void DeleteBusPublishInfo(string busName)
        {
            throw new NotImplementedException();
        }

        public void DeleteBusSubscribeInfo(string busName)
        {
            throw new NotImplementedException();
        }

        public void UploadPublishInfo(List<PublishInfo> info)
        {
            // ZooKeeper Node Template
            // PublishInfo Path:/EasyBus/PublishInfo/{PubBusName}/{RouteKey}/{ServerNode}
            // SubscribeInfo Path:/EasyBus/SubscribeInfo/{PubBusName}/{RouteKey}/{SubBusName}/{ServerNode}
            // ListenerInfo Path:/EasyBus/ListenerInfo/{SubBusName}/{ServerNode}/Listerer-{Instacne's HashCode}
             
            // TODO:以上のコードを以下のコードと取り替えるつもりだ
            // PublishInfo Path:/EasyBus/PublishInfo/{PubBusName}/{RouteKey}/{ServerNode}
            // SubscribeInfo Path:/EasyBus/SubscribeInfo/{PubBusName}/{RouteKey}/{SubBusNama}-{NodeName}-{SubscribeName}
            // ListenerInfo Path:/EasyBus/ListenerInfo/{SubBusName}/{ServerNode}/{ListenerName}

            info.ForEach(v =>
            {
                string busNameNodePath = GetPublishBusNameNodePath(v.BusName);
                string routeKeyNodePath = GetPublishRouteKeyNodePath(v.BusName, v.RouteKey);
                string serverNodePath = GetPublishServerNodePath(v.BusName, v.RouteKey, v.NodeName);
                 
                // 发布端建立路由关键字节点，总线节点，服务器节点
                this.TryCreateZNode(busNameNodePath, null, ZK.CreateMode.Persistent);
                this.TryCreateZNode(routeKeyNodePath, null, ZK.CreateMode.Persistent);
                this.TryCreateZNode(serverNodePath, JsonConvert.SerializeObject(v).ToBytesByUTF8(), ZK.CreateMode.Ephemeral);

                // 在订阅端信息加入路由关键字节点
                string subscribe_publishBusPath = GetSubscribePubBusNameNodePath(v.BusName);
                string subscribe_routeKeyPath = GetSubscribeRouteKeyNodePath(v.BusName, v.RouteKey);

                this.TryCreateZNode(subscribe_publishBusPath, null, ZK.CreateMode.Persistent);
                this.TryCreateZNode(subscribe_routeKeyPath, null, ZK.CreateMode.Persistent);
            });
        }

        public void UploadSubscribeInfo(List<SubscribeInfo> info)
        {
            info.ForEach(v => 
            {
                string publishBusPath = GetSubscribePubBusNameNodePath(v.PubBusName);
                string routeKeyPath = GetSubscribeRouteKeyNodePath(v.PubBusName, v.RouteKey);

                // 检查要订阅的路由关键字是否存在
                (this._zk.Exists(routeKeyPath, false) == null).WhenTrue(() => { throw new EasyBusException("BusName{0}，RouteKey:{1}，不存在。".Fmt(v.PubBusName, v.RouteKey)); });

                string subscribeNodePath = GetSubscribeNameNodePath(v.PubBusName, v.RouteKey, "{0}-{1}-{2}".Fmt(v.BusName, v.NodeName, v.Name));

                // 创建订阅端节点信息
                this.TryCreateZNode(subscribeNodePath, JsonConvert.SerializeObject(v).ToBytesByUTF8(), ZK.CreateMode.Ephemeral);
            });
        }

        /// <summary>
        /// 上传总线的监听信息
        /// </summary>
        /// <param name="info">监听信息集合</param>
        public void UploadListenerInfo(List<ListenerInfo> info) 
        {
            info.ForEach(v =>
            {
                string subBusNamePath = GetListenerSubBusNameNodePath(v.SubBusName);
                string serverNodePath = GetListenerServerNodePath(v.SubBusName, v.NodeName);
                string listenerNodePath = GetListenerNameNodePath(v.SubBusName, v.NodeName, v.Name);

                this.TryCreateZNode(subBusNamePath, null, ZK.CreateMode.Persistent);
                this.TryCreateZNode(serverNodePath, null, ZK.CreateMode.Persistent);
                this.TryCreateZNode(listenerNodePath, JsonConvert.SerializeObject(v).ToBytesByUTF8(), ZK.CreateMode.Ephemeral);
            });
        }

        #endregion

        /// <summary>
        /// 获取 ZK 对象
        /// </summary>
        /// <returns></returns>
        protected virtual ZK.ZooKeeper GetZK(TimeSpan sessionTimeout) 
        {
            ZK.ZooKeeper zoo = new ZK.ZooKeeper(this._connectionString, sessionTimeout, new SessionChangeWatcher(this._bus));
            
            zoo.WaitToConnectionCompleted(new TimeSpan(0, 1, 0));

            return zoo;
        }

        protected virtual void InitZooKeeperNode() 
        {
            // 检查 ROOT 节点
            this.TryCreateZNode(ROOT, null, ZK.CreateMode.Persistent);

            // 发布
            this.TryCreateZNode(PUBLISH, null, ZK.CreateMode.Persistent);

            // 订阅
            this.TryCreateZNode(SUBSCRIBE, null, ZK.CreateMode.Persistent);

            // 监听
            this.TryCreateZNode(LISTENER, null, ZK.CreateMode.Persistent);
        }

        protected virtual void TryCreateZNode(string path, byte[] data, ZK.CreateMode createMode) 
        {
            (this._zk.Exists(path, false) == null).WhenTrue(() => 
            {
                try
                {
                    this._zk.Create(path, data, ZK.Ids.OPEN_ACL_UNSAFE, createMode);
                }
                catch (ZK.KeeperException.NodeExistsException) { }
                catch (Exception ce) 
                {
                    throw new EasyBusException("初始化节点：{0}失败，详细见内部异常".Fmt(path), ce);
                }
            });
        }

        public static string GetPublishBusNameNodePath(string busName) 
        {
            return PUBLISH_BUSNAME_NODE.Replace(PUBBUSNAME_PREFIX, busName);
        }

        public static string GetPublishRouteKeyNodePath(string busName, string routeKey) 
        {
            return PUBLISH_ROUTEKEY_NODE.Replace(PUBBUSNAME_PREFIX, busName)
                                        .Replace(ROUTEKEY_PREFIX, routeKey);
        }

        public static string GetPublishServerNodePath(string busName, string routeKey, string nodeName) 
        {
            return PUBLISH_SERVER_NODE.Replace(PUBBUSNAME_PREFIX, busName)
                                      .Replace(ROUTEKEY_PREFIX, routeKey)
                                      .Replace(NODENAME_PREFIX, nodeName);
        }

        public static string GetSubscribePubBusNameNodePath(string pubBusName) 
        {
            return SUBSCRIBE_PUBBUSNAME_NODE.Replace(PUBBUSNAME_PREFIX, pubBusName);
        }

        public static string GetSubscribeRouteKeyNodePath(string pubBusName, string routeKey) 
        {
            return SUBSCRIBE_ROUTEKEY_NODE.Replace(PUBBUSNAME_PREFIX, pubBusName)
                                          .Replace(ROUTEKEY_PREFIX, routeKey);
        }

        public static string GetSubscribeNameNodePath(string pubBusName, string routeKey, string subscribeName) 
        {
            return SUBSCRIBE_NODE.Replace(PUBBUSNAME_PREFIX, pubBusName)
                                 .Replace(ROUTEKEY_PREFIX, routeKey)
                                 .Replace(SUBSCRIBE_NAME_PREFIX, subscribeName);
        }

        public static string GetListenerSubBusNameNodePath(string subBusName) 
        {
            return LISTENER_SUBBUSNAME_NODE.Replace(SUBBUSNAME_PREFIX, subBusName);
        }

        public static string GetListenerServerNodePath(string subBusName, string nodeName)
        {
            return LISTENER_SERVER_NODE.Replace(SUBBUSNAME_PREFIX, subBusName)
                                       .Replace(NODENAME_PREFIX, nodeName);
        }

        public static string GetListenerNameNodePath(string subBusName, string nodeName, string listenerName) 
        {
            return LISTENER_NAME_NODE.Replace(SUBBUSNAME_PREFIX, subBusName)
                                     .Replace(NODENAME_PREFIX, nodeName)
                                     .Replace(LISTENER_NAME_PREFIX, listenerName);
        }

        #region IDisposable 成员

        public void Dispose()
        {
            if (this._zk != null) 
            {
                this._zk.Dispose();
            }
        }

        #endregion
    }
}
